package com.atguigu.flink.chapter05.Sink;

import com.atguigu.flink.bean.WaterSensor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;

/**
 * TODO
 *
 * @author cjp
 * @version 1.0
 * @date 2021/8/10 8:59
 */
public class Flink02_Redis {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<String> socketDS = env.socketTextStream("localhost", 9999);

        SingleOutputStreamOperator<WaterSensor> sensorDS = socketDS.map((MapFunction<String, WaterSensor>) value -> {
            String[] datas = value.split(",");
            return new WaterSensor(
                    datas[0],
                    Long.parseLong(datas[1]),
                    Integer.parseInt(datas[2])
            );
        });

        //TODO Sink - Redis
        FlinkJedisPoolConfig jedisPoolConfig = new FlinkJedisPoolConfig.Builder()
                .setHost("hadoop1")
                .setPort(6379)
                .build();

        RedisSink<WaterSensor> redisSink = new RedisSink<>(jedisPoolConfig,
                new RedisMapper<WaterSensor>() {
                    @Override
                    public RedisCommandDescription getCommandDescription() {
                        RedisCommandDescription redisCommandDescription = new RedisCommandDescription(RedisCommand.HSET, "flink210323");
                        return redisCommandDescription;
                    }

                    /**
                     * 从数据里面提取key： 如果是hash结构，提取的是 hash 的 key
                     * @param data
                     * @return
                     */
                    @Override
                    public String getKeyFromData(WaterSensor data) {
                        return data.getId();
                    }

                    /**
                     * 从数据里面提取 value： 如果是hash结构，提取的是 hash 的 value
                     * @param data
                     * @return
                     */
                    @Override
                    public String getValueFromData(WaterSensor data) {
                        return data.getVc().toString();
                    }
                });

        sensorDS.addSink(redisSink);

        env.execute();
    }
}
/*

 */